package de.fhkoeln.avs.parallelbreadthfirstsearch;

import com.infinitegraph.ConfigurationException;
import com.infinitegraph.StorageException;
import de.fhkoeln.avs.graphbuild.infinitegraph.storage.AvsVertex;
import de.fhkoeln.avs.parallelbreadthfirstsearch.infinitegraph.CacheBuilder;
import de.fhkoeln.avs.parallelbreadthfirstsearch.infinitegraph.Crawler;
import de.fhkoeln.avs.parallelbreadthfirstsearch.infinitegraph.DatabaseConnection;
import de.fhkoeln.avs.parallelbreadthfirstsearch.infinitegraph.LocalThreadCrawler;
import de.fhkoeln.avs.parallelbreadthfirstsearch.infinitegraph.PFBSCallback;
import de.fhkoeln.avs.strategypattern.ISearchStrategy;
import java.lang.reflect.Constructor;
import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.Semaphore;
import java.util.logging.Level;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 *
 * @author Markus and Muhammed
 */
public class StrategyParallelBreadthFirstForInfiniteGraph implements ISearchStrategy {

    private static Logger logger = LoggerFactory.getLogger(StrategyParallelBreadthFirstForInfiniteGraph.class);
    private static Integer CRAWLERNUMBER = 4;
    private final BlockingQueue<AvsVertex> fs = new LinkedBlockingQueue<AvsVertex>();
    private final Map<String, Integer> distances = new ConcurrentHashMap<String, Integer>();

    public int search(String startTerm, String endTerm) {
        try {
            DatabaseConnection.getConnection();
        } catch (StorageException ex) {
            logger.error("StorageException. Cant build graph \n"
                    + ex.getMessage());
            return -1;
        } catch (ConfigurationException ex) {
            logger.error("ConfigurationException. Cant build graph \n"
                    + ex.getMessage());
            return -1;
        }
//		DatabaseConnection.getTransaction();
        Map<String, Integer> distances = null;
        try {
            distances = breathFirstSearch(startTerm);
        } catch (Exception ex) {
            java.util.logging.Logger.getLogger(StrategyParallelBreadthFirstForInfiniteGraph.class.getName()).log(Level.SEVERE, null, ex);
        }
        if (distances != null) {
            logger.info("Distanz ist" + distances.get(endTerm));
            return distances.get(startTerm);
        }

        return -1;
    }

    private Map<String, Integer> breathFirstSearch(
            String startTerm) throws Exception {
        //Initialisierungen
        BigDecimal level = new BigDecimal(1);

        final List<Crawler> crawlers = createCrawler(CRAWLERNUMBER);

        //Cache the graph because of performance issues
//        Stopwatch stopwatch = new Stopwatch();
//        stopwatch.start();
        AvsVertex startVertex = CacheBuilder.cacheGraphAndReturnStart(startTerm);
//        stopwatch.stop();
//        logger.info("Graph cached in: " + stopwatch.getElapsedTime() + "ms.");
//		
//		
//        stopwatch.reset();
//        stopwatch.start();
        fs.add(startVertex);

        distances.put(startTerm, 0);
        ExecutorService executor = Executors.newFixedThreadPool(CRAWLERNUMBER);

        //--------------------------------
        while (!fs.isEmpty()) {
            logger.info("Durchlauf: " + level.intValue() + ". Now in Stack: " + fs.size());
            Integer currentCrawler = 0;
            final BigDecimal semaphoreCount = new BigDecimal(CRAWLERNUMBER < fs.size() ? CRAWLERNUMBER : fs.size());
            final Semaphore semCrawler = new Semaphore(semaphoreCount.intValue());
            final Semaphore semNew = new Semaphore(CRAWLERNUMBER);
            final PFBSCallback finishedThreadCallback = getThreadFinishedCallback(semCrawler, semNew, level.intValue());

            //Setze den Callback mit den aktuellen Semaphore Werten f�r alle Crawler
            setCallbackForCrawlers(crawlers, finishedThreadCallback);
            do {
                sendVerticesAndFormTBufAsync(executor, currentCrawler, semaphoreCount, semCrawler, crawlers);
                currentCrawler++;
                if (CRAWLERNUMBER == currentCrawler) {
                    currentCrawler = 0;
                }
            } while (!fs.isEmpty());
            semCrawler.acquireUninterruptibly(semaphoreCount.intValue());

//			//---SendTBufAufrufen und warten bis alle TBufs verschickt und von der Gegenseite
//			//verarbeitet wurden
            callSendTBufAsync(executor, semaphoreCount, semNew, crawlers);
            semNew.acquireUninterruptibly(CRAWLERNUMBER);
//			
//			//Alles abgeschlossen, Neues FS aufgebaut
//			//Crawler wieder zur�cksetzen
            cleanCrawler(crawlers);
//			
//			//Level f�r den n�chsten Durchlauf erh�hen
            level = level.add(new BigDecimal(1));
        }
        //Stoppe alle Crawler
        stopCrawlers(crawlers);

//        stopwatch.stop();
//        logger.info("Graph traversed in: " + stopwatch.getElapsedTime() + "ms.");
        return distances;
    }

    private static List<Crawler> createCrawler(Integer crawlerNumber) throws Exception {
        List<Crawler> crawlers = new ArrayList<Crawler>();
        for (Integer i = 1; i <= crawlerNumber; i++) {
            Class c = Class.forName(LocalThreadCrawler.class.getName());
            Constructor con = c.getConstructor(String.class);
            Crawler crawler = (Crawler) con.newInstance(i.toString());//With RMI other identifier
            crawler.startCrawler();
            crawlers.add(crawler);
        }
        for (Crawler crawler : crawlers) {
            crawler.setCrawlers(crawlers);
        }
        return crawlers;
    }

    private PFBSCallback getThreadFinishedCallback(final Semaphore semCrawler, final Semaphore semNew, final Integer level) {

        PFBSCallback finishedThreadCallback = new PFBSCallback() {
            @Override
            public void sendTBufsFinished(Map<String, AvsVertex> distanceMap) {
                for (AvsVertex vertex : distanceMap.values()) {
                    if (distances.get(vertex.getName()) == null) {
                        distances.put(vertex.getName(), level);
                        fs.add(vertex);
                    }
                }
                semNew.release();
            }

            @Override
            public void buildTBufsFinished() {
                semCrawler.release();
            }
        };
        return finishedThreadCallback;
    }

    private void callSendTBufAsync(ExecutorService executor,
            final BigDecimal semaphoreCount, final Semaphore semNew,
            final List<Crawler> crawlers) throws InterruptedException {
        for (final Crawler crawler : crawlers) {
            semNew.acquire();
            executor.submit(new Runnable() {
                @Override
                public void run() {
                    try {
                        crawler.sendTBufs(semaphoreCount.intValue());
                    } catch (InterruptedException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                }
            });
        }
    }

    private void cleanCrawler(final List<Crawler> crawlers) {
        for (Crawler crawler : crawlers) {
            crawler.cleanCrawler();
        }
    }

    private void sendVerticesAndFormTBufAsync(ExecutorService executor,
            Integer currentCrawler, final BigDecimal semaphoreCount,
            final Semaphore semCrawler, final List<Crawler> crawlers)
            throws InterruptedException {
        semCrawler.acquire();
        final Crawler crawler = crawlers.get(currentCrawler);
        final AvsVertex currentVertex = fs.poll();
        executor.submit(new Runnable() {
            @Override
            public void run() {
                try {
                    crawler.distributeChildVertices(currentVertex); //semaphoreCount.intValue());
                } catch (Exception e) {
                    e.printStackTrace();
                    throw new RuntimeException("Algorithmus aufgrund eines Fehlers abgebrochen");
                }
            }
        });
    }

    private void setCallbackForCrawlers(final List<Crawler> crawlers,
            final PFBSCallback finishedThreadCallback) {
        for (Crawler crawler : crawlers) {
            crawler.setCallback(finishedThreadCallback);
        }
    }

    private void stopCrawlers(final List<Crawler> crawlers) {
        for (Crawler crawler : crawlers) {
            crawler.stopCrawler();
        }
    }
}
